package com.dai.rpc.netty.client;

import com.dai.rpc.config.MyRpcConfig;
import com.dai.rpc.constant.CompressTypeEnum;
import com.dai.rpc.constant.MessageTypeEnum;
import com.dai.rpc.constant.SerializationTypeEnum;
import com.dai.rpc.exceptions.MyRpcException;
import com.dai.rpc.factory.SingletonFactory;
import com.dai.rpc.loadbanlance.LoadBalance;
import com.dai.rpc.message.RpcMessage;
import com.dai.rpc.message.RpcRequest;
import com.dai.rpc.message.RpcResponse;
import com.dai.rpc.netty.cache.ChannelCache;
import com.dai.rpc.netty.codec.MyRpcDecoder;
import com.dai.rpc.netty.codec.MyRpcEncoder;
import com.dai.rpc.netty.handler.MyNettyClientHandler;
import com.dai.rpc.netty.handler.idle.ConnectionWatchdog;
import com.dai.rpc.netty.task.UpdateServiceTask;
import com.dai.rpc.register.zookeeper.ZKServiceDiscovery;
import com.dai.rpc.util.SPIUtils;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timer;
import lombok.extern.slf4j.Slf4j;
import java.net.InetSocketAddress;
import java.util.Optional;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

@Slf4j
public class NettyClient implements MyClient {

    private final Bootstrap bootstrap;
    private final EventLoopGroup eventLoopGroup;
    private final UnprocessedRequests unprocessedRequests;
    private final ZKServiceDiscovery zkServiceDiscovery;
    private MyRpcConfig myRpcConfig;
    // 线程安全集合
    private final Set<String> SERVICES = new CopyOnWriteArraySet<>();
    private Timer timer = new HashedWheelTimer();
    private ChannelCache channelCache;
    private Timer serviceTimer;
    private LoadBalance loadBalance;

    public void setMyRpcConfig(MyRpcConfig myRpcConfig) {
        this.myRpcConfig = myRpcConfig;
        // 获得配置类后，在获取配置类中的实例
        this.loadBalance = SPIUtils.loadBalance(myRpcConfig.getLoadBalancer());
    }


    public NettyClient() {
        this.eventLoopGroup = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        this.unprocessedRequests = SingletonFactory.getInstance(UnprocessedRequests.class);
        zkServiceDiscovery = SingletonFactory.getInstance(ZKServiceDiscovery.class);
        channelCache = SingletonFactory.getInstance(ChannelCache.class);

        bootstrap.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new LoggingHandler(LogLevel.INFO))
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000);
                /*.handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        // 添加心跳检测,3s 没有接收到写请求，就发送心跳检测，告诉服务端我还活着，顺便获取服务端状态
                        socketChannel.pipeline().addLast("heart",new IdleStateHandler(0,3,0, TimeUnit.SECONDS));
                        // 添加编解码器
                        socketChannel.pipeline().addLast("decoder",new MyRpcDecoder());
                        socketChannel.pipeline().addLast("encoder",new MyRpcEncoder());
                        // 添加消息处理器
                        socketChannel.pipeline().addLast("handler",new MyNettyClientHandler());
                    }
                });*/
    }

    @Override
    public Object sendRequest(RpcRequest rpcRequest) {


        InetSocketAddress inetAddress = null;
        CompletableFuture<RpcResponse<Object>> resultFuture = new CompletableFuture<>();
        // 从缓存中查看是否有服务提供方的地址
        String ipPort = null;

        // 使用负载均衡策略从本地获取可用实例信息
        inetAddress = loadBalance.loadBalance(SERVICES);

        if(inetAddress == null){
            // 1.获取连接
            //InetSocketAddress inetAddress = new InetSocketAddress(host, port);
            try{
                // 从zookeeper获取对应服务的地址
                inetAddress = zkServiceDiscovery.lookupService(rpcRequest);
                log.info("获取到了新的实例：{}",inetAddress.toString());
                // 缓存实例
                ipPort = inetAddress.getHostName() + "," + inetAddress.getPort();
                SERVICES.add(ipPort);
                // 触发定时任务，从ZK中获取新的实例
                if(serviceTimer == null){
                    serviceTimer = new HashedWheelTimer();
                    serviceTimer.newTimeout(new UpdateServiceTask(SERVICES, zkServiceDiscovery,rpcRequest),10,TimeUnit.SECONDS);
                }

            }catch (Exception e){
                e.printStackTrace();
                // 任务标识为有异常
                log.info("客户端获取 zookeeper 实例出错");
                // 返回结果
                resultFuture.completeExceptionally(e.getCause());
                throw new MyRpcException(e.getMessage());
            }

        }
        CompletableFuture<Channel> channelCompletableFuture = new CompletableFuture<>();

        // 链路检测狗
        ConnectionWatchdog watchdog = new ConnectionWatchdog(bootstrap,channelCompletableFuture,timer,inetAddress,true,channelCache) {
            @Override
            public ChannelHandler[] channelHandlers() {
                return new ChannelHandler[]{
                        this,
                        new IdleStateHandler(0,3,0, TimeUnit.SECONDS),
                        new MyRpcDecoder(),
                        new MyRpcEncoder(),
                        new MyNettyClientHandler()
                };
            }

            @Override
            public void clear(InetSocketAddress inetAddress) {
                log.info("链路检测狗触发：删除provider服务缓存成功...");
                SERVICES.remove(inetAddress.getHostName()+","+inetAddress.getPort());
                // 删除连接
                channelCache.remove(inetAddress);
            }
        };
        // 添加链路检测狗的处理器
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                socketChannel.pipeline().addLast(watchdog.channelHandlers());
            }
        });


        try {
            // 阻塞获取连接
            Channel channel = getChannel(inetAddress, channelCompletableFuture);
            // Channel channel = channelCompletableFuture.get();
            if(channel.isActive()){
                unprocessedRequests.put(rpcRequest.getRequestId(),resultFuture);
                // 构建发送的对象
                RpcMessage rpcMessage = RpcMessage.builder()
                        .messageType(MessageTypeEnum.REQUEST.getCode())
                        .codec(SerializationTypeEnum.KRYO.getCode())
                        .compress(CompressTypeEnum.GZIP.getCode())
                        .data(rpcRequest)
                        .build();
                // 发送数据
                channel.writeAndFlush(rpcMessage).addListener(new ChannelFutureListener() {

                    @Override
                    public void operationComplete(ChannelFuture channelFuture) throws Exception {
                        if(channelFuture.isSuccess()){
                            // 代表连接成功，将channel放入任务中
                            log.info("请求成功！");
                        }else{
                            //发送数据失败
                            channelFuture.channel().close();
                            //任务标识为有异常
                            resultFuture.completeExceptionally(channelFuture.cause());
                            throw new MyRpcException("连接服务器失败");
                        }
                    }
                });
            }
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }

        return resultFuture;
    }

    private Channel getChannel(InetSocketAddress inetAddress, CompletableFuture<Channel> channelCompletableFuture) throws ExecutionException, InterruptedException {

        // 先从缓存中获取channel
        Channel channel = channelCache.get(inetAddress);
        if(channel == null){
            // 如果没有则从连接中获取channel
            channel = doConnect(inetAddress,channelCompletableFuture);
            // 放入缓存
            channelCache.set(inetAddress,channel);
        }else{
            log.info("从缓存中获取channel...");
        }

        return channel;

    }

    private Channel doConnect(InetSocketAddress inetAddress, CompletableFuture<Channel> channelCompletableFuture) throws ExecutionException, InterruptedException {

        bootstrap.connect(inetAddress).addListener(new ChannelFutureListener() {

            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                if(channelFuture.isSuccess()){
                    // 代表连接成功，将channel放入任务中
                    channelCompletableFuture.complete(channelFuture.channel());
                }else{
                    // 连接失败，删除缓存
                    log.info("删除provider缓存成功....");
                    SERVICES.remove(inetAddress.getHostName() + "," + inetAddress.getPort());
                    channelCompletableFuture.completeExceptionally(channelFuture.cause());
                    throw new MyRpcException("连接服务器失败");
                }
            }
        });

        // 获取连接的过程是阻塞的，通过异步对象获取
        return channelCompletableFuture.get();
    }
}
